use mio::{Interest, Poll};
use std::{
    io::{self, ErrorKind},
    thread,
    time::Duration,
};
use tokio::{sync::oneshot::error, time::Instant};

use crate::{
    availability::Availability,
    builder,
    handle::ServerHandle,
    socket::MioListener,
    waker_queue::{WakerQueue, WAKER_TOKEN},
    worker::{Conn, ServerWorker, WorkerHandleAccept, WorkerHandleServer},
    ServerBuilder,
};

const TIMEOUT_DURATION_ON_ERROR: Duration = Duration::from_millis(510);

struct ServerSocketInfo {
    token: usize,

    lst: MioListener,

    /// Timeout is used to mark the deadline when this socket's listener should be registered again
    /// after an error.
    timeout: Option<Instant>,
}
pub(crate) struct Accept {
    poll: Poll,
    waker_queue: WakerQueue,
    handles: Vec<WorkerHandleAccept>,
    srv: ServerHandle,
    next: usize,
    avail: Availability,
    /// use the smallest duration from sockets timeout.
    timeout: Option<Duration>,
    paused: bool,
}

impl Accept {
    pub(crate) fn start(
        sockets: Vec<(usize, MioListener)>,
        builder: &ServerBuilder,
    ) -> io::Result<(WakerQueue, Vec<WorkerHandleServer>, thread::JoinHandle<()>)> {
        let handle_server = ServerHandle::new(builder.cmd_tx.clone());

        // 构建poll实例和 waker
        let poll = Poll::new()?;
        // waker 队列
        let waker_queue = WakerQueue::new(poll.registry())?;
        // start workers and collect handles
        let (handles_accept, handles_server) = (0..builder.threads)
            .map(|idx| {
                // clone service factories
                let factories = builder
                    .factories
                    .iter()
                    .map(|f| f.clone_factory())
                    .collect::<Vec<_>>();
                // start worker using service factories
                ServerWorker::start(idx, factories, waker_queue.clone(), builder.worker_config)
            })
            .collect::<io::Result<Vec<_>>>()?
            .into_iter()
            .unzip();
        let (mut accept, mut sockets) = Accept::new_with_sockets(
            poll,
            waker_queue.clone(),
            sockets,
            handles_accept,
            handle_server,
        )?;

        let accept_handle = thread::Builder::new()
            .name("actix-server acceptor".to_owned())
            .spawn(move || accept.poll_with(&mut sockets))
            .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?;

        Ok((waker_queue, handles_server, accept_handle))
    }

    fn new_with_sockets(
        poll: Poll,
        waker_queue: WakerQueue,
        sockets: Vec<(usize, MioListener)>,
        accept_handles: Vec<WorkerHandleAccept>,
        server_handle: ServerHandle,
    ) -> io::Result<(Accept, Box<[ServerSocketInfo]>)> {
        let sockets = sockets
            .into_iter()
            .map(|(token, mut lst)| {
                poll.registry()
                    .register(&mut lst, mio::Token(token), Interest::READABLE)?;
                Ok(ServerSocketInfo {
                    token,
                    lst,
                    timeout: None,
                })
            })
            .collect::<io::Result<_>>()?;

        let mut avail = Availability::default();

        // Assume all handles are avail at construct time.
        avail.set_available_all(&accept_handles);

        let accept = Accept {
            poll,
            waker_queue,
            handles: accept_handles,
            srv: server_handle,
            next: 0,
            avail,
            timeout: None,
            paused: false,
        };

        Ok((accept, sockets))
    }

    fn poll_with(&mut self, sokets: &mut [ServerSocketInfo]) {
        let mut events = mio::Events::with_capacity(256);

        loop {
            if let Err(err) = self.poll.poll(&mut events, self.timeout) {
                match err.kind() {
                    io::ErrorKind::Interrupted => {}
                    _ => panic!("Poll error: {}", err),
                }
            }

            // 处理接收到的事件
            for event in events.iter() {
                let token = event.token();
                match token {
                    WAKER_TOKEN => {}
                    _ => {
                        let token = usize::from(token);
                    }
                }
            }
        }
    }

    fn set_timeout(&mut self, duration: Duration) {
        match self.timeout {
            Some(ref mut timeout) => {
                if *timeout > duration {
                    // 仅更新Some<T> 的 T
                    *timeout = duration;
                }
            }
            None => self.timeout = Some(duration),
        }
    }

    fn register(&self, info: &mut ServerSocketInfo) -> io::Result<()> {
        // On windows, calling register without deregister cause an error.
        // See https://github.com/actix/actix-web/issues/905
        // Calling reregister seems to fix the issue.
        let token = mio::Token(info.token);
        self.poll
            .registry()
            .register(&mut info.lst, token, Interest::READABLE)
            .or_else(|_| {
                self.poll
                    .registry()
                    .reregister(&mut info.lst, token, Interest::READABLE)
            })
    }

    fn register_logged(&self, info: &mut ServerSocketInfo) {
        match self.register(info) {
            Ok(_) => println!("resume accepting connections on {}", info.lst.local_addr()),
            Err(err) => println!("can not register server socket {}", err),
        }
    }

    fn deregister_logged(&self, info: &mut ServerSocketInfo) {
        match self.poll.registry().deregister(&mut info.lst) {
            Ok(_) => println!("paused accepting connections on {}", info.lst.local_addr()),
            Err(err) => println!("can not deregister server socket {}", err),
        }
    }

    fn deregister_all(&self, sockets: &mut [ServerSocketInfo]) {
        // This is a best effort implementation with following limitation:
        //
        // Every ServerSocketInfo with associated timeout will be skipped and it's timeout is
        // removed in the process.
        //
        // Therefore WakerInterest::Pause followed by WakerInterest::Resume in a very short gap
        // (less than 500ms) would cause all timing out ServerSocketInfos be re-registered before
        // expected timing.
        sockets
            .iter_mut()
            // Take all timeout.
            // This is to prevent Accept::process_timer method re-register a socket afterwards.
            .map(|info| (info.timeout.take(), info))
            // Socket info with a timeout is already deregistered so skip them.
            .filter(|(timeout, _)| timeout.is_none())
            .for_each(|(_, info)| self.deregister_logged(info));
    }

    fn accept_one(&mut self, mut conn: Conn) {
        loop {
            let next = self.next();
            let idx = next.idx();

            if self.avail.get_available(idx) {
                match self.send_connection(conn) {
                    Ok(_) => return,
                    Err(c) => conn = c,
                }
            } else {
                //TODO: 为什么再设置一次？
                self.avail.set_available(idx, false);
                self.set_next();

                // 如果都不可用
                if !self.avail.available() {
                    // TODO:(待观察)这里发送直至成功
                    while let Err(c) = self.send_connection(conn) {
                        conn = c;
                    }
                    return;
                }
            }
        }
    }

    fn accept(&mut self, sockets: &mut [ServerSocketInfo], token: usize) {
        while self.avail.available() {
            let info = &mut sockets[token];

            match info.lst.accept() {
                Ok(io) => {
                    let conn = Conn { io, token };
                    self.accept_one(conn);
                }
                Err(ref err) if err.kind() == ErrorKind::WouldBlock => return,
                Err(ref err) if connection_error(err) => continue,
                Err(err) => {
                    println!("error accepting connection: {}", err);

                    // deregister listener temporary
                    self.deregister_logged(info);

                    // sleep after error. write the timeout to socket info as later
                    // the poll would need it mark which socket and when it's
                    // listener should be registered
                    info.timeout = Some(Instant::now() + Duration::from_millis(500));
                    self.set_timeout(TIMEOUT_DURATION_ON_ERROR);
                }
            }
        }
    }

    fn send_connection(&mut self, conn: Conn) -> Result<(), Conn> {
        let next = self.next();
        match next.send(conn) {
            Ok(_) => {
                // 累加workerHandle的计数器
                // 当到达最大值时，设置不可用
                if !next.inc_counter() {
                    let idx = next.idx();
                    self.avail.set_available(idx, false);
                }
                self.set_next();
                Ok(())
            }
            Err(conn) => {
                // 发送失败则移除next的handle
                self.remove_next();
                // All workers are gone and Conn is nowhere to be sent.
                // Treat this situation as Ok and drop Conn.
                if self.handles.is_empty() {
                    // TODO: trace
                    println!("no workers");
                    return Ok(());
                } else if self.handles.len() <= self.next {
                    // 避免next指向无效位置
                    self.next = 0;
                }
                Err(conn)
            }
        }
    }

    fn next(&self) -> &WorkerHandleAccept {
        &self.handles[self.next]
    }

    fn set_next(&mut self) {
        self.next = (self.next + 1) % self.handles.len()
    }

    fn remove_next(&mut self) {
        let handle = self.handles.swap_remove(self.next);
        let idx = handle.idx();
        // 发送失败消息
        self.srv.worker_faulted(idx);
        //TODO: 这里设置为false的目的是？
        self.avail.set_available(idx, false);
    }
}

fn connection_error(e: &io::Error) -> bool {
    matches!(
        e.kind(),
        ErrorKind::ConnectionRefused | ErrorKind::ConnectionAborted | ErrorKind::ConnectionReset
    )
}
